//
// Created by huyi on 24-7-12.
//
#include "AmaInstance.h"

DECLARE_string(ama_server_ip1);
DECLARE_string(ama_server_ip2);
DECLARE_int32(ama_server_port1);
DECLARE_int32(ama_server_port2);
DECLARE_string(ama_username);
DECLARE_string(ama_password);
DECLARE_int32(print_period);

namespace tradev::app {

int AmaInstance::open() {
  auto username = FLAGS_ama_username;
  auto password = FLAGS_ama_password;
  auto server1  = FLAGS_ama_server_ip1;
  auto server2  = FLAGS_ama_server_ip2;
  auto port1    = FLAGS_ama_server_port1;
  auto port2    = FLAGS_ama_server_port2;

  XLOG(INFO) << "AMA username " << username;
  XLOG(INFO) << "AMA password " << password;
  XLOG(INFO) << "AMA server1 " << server1;
  XLOG(INFO) << "AMA server2 " << server2;
  XLOG(INFO) << "AMA port1 " << port1;
  XLOG(INFO) << "AMA port2 " << port2;

  //
  Cfg cfg; // 准备AMA配置

  // 通道模式设置及各个通道说明:
  // cfg.channel_mode = amd::ama::ChannelMode::kTCP;   ///< TCP 方式计入上游行情系统
  // cfg.channel_mode = amd::ama::ChannelMode::kAMI;   ///< AMI 组播方式接入上游行情系统
  // cfg.channel_mode = amd::ama::ChannelMode::kRDMA;  ///< 开启硬件加速RDMA通道,抓取网卡数据包数据
  // cfg.channel_mode = amd::ama::ChannelMode::kEXA;   ///< 开启硬件加速EXA通道,抓取网卡数据包数据
  // cfg.channel_mode = amd::ama::ChannelMode::kPCAP;  ///< 开启硬件加速PCAP通道,抓取网卡数据包数据
  // cfg.channel_mode = amd::ama::ChannelMode::kMDDP;  ///< 直接接入交易所网关组播数据，现在只有深圳交易所开通了此服务
  // cfg.channel_mode = amd::ama::ChannelMode::kTCP|amd::ama::ChannelMode::kAMI;  ///< 同时通过TCP方式和AMI组播方式接入上游，通过cfg.ha_mode 设置对应的高可用设置模式
  cfg.channel_mode = amd::ama::ChannelMode::kTCP;

  //TCP传输数据方式: 0 不压缩 1 华锐自定义压缩 2 zstd压缩(仅TCP模式有效)
  cfg.tcp_compress_mode = 0U;

  // 通道高可用模式设置
  // 1. cfg.channel_mode 为单通道时,建议设置值为kMasterSlaveA / kMasterSlaveB
  // 2. cfg.channel_mode 混合开启多个通道时,根据需求设置不同的值
  //  1) 如果需要多个通道为多活模式接入,请设置kRegularDataFilter值
  //  2) 如果需要多个通道互为主备接入，请设置值为kMasterSlaveA / kMasterSlaveB,kMasterSlaveA / kMasterSlaveB 差别请参看注释说明
  //     通道优先级从高到低依次为 kRDMA/kEXA/kMDDP/kAMI/kTCP/kPCAP
  cfg.ha_mode = amd::ama::HighAvailableMode::kMasterSlaveA;

  // 设置日志最小级别：Info级, AMA内部日志通过 OnLog 回调函数返回
  cfg.min_log_level = amd::ama::LogLevel::kInfo;

  // 设置是否输出监控数据: true(是), false(否), 监控数据通过OnIndicator 回调函数返回
  // 监控数据格式为json, 主要监控数据包括订阅信息，数据接收数量统计
  // 数据量统计:包括接收数量和成功下发的数量统计,两者差值为过滤的数据量统计
  //   eg: "RecvSnapshot": "5926", "SuccessSnapshot": "5925",表示接收了快照数据5926个,成功下发5925个，过滤数据为 5926 - 5925 = 1 个
  // 过滤的数据有可能为重复数据或者非订阅数据
  cfg.is_output_mon_data = false;

  // 设置逐笔保序开关: true(开启保序功能) , false(关闭保序功能)
  // 主要校验逐笔成交数据和逐笔委托数据是否有丢失,如果丢失会有告警日志,缓存逐笔数据并等待keep_order_timeout(单位s)时间等待上游数据重传,
  // 如果超过此时间,直接下发缓存数据,默认数据已经丢失,如果之后丢失数据过来也会丢弃。
  // 同时由于深圳和上海交易所都是通道内序号连续,如果打开了保序开关,必须订阅全部代码的逐笔数据,否则一部分序号会被订阅过滤,导致数据超时等待以及丢失告警。  cfg.keep_order            = false;
  cfg.keep_order_timeout_ms = 3000U;

  //设置默认订阅: true 代表默认全部订阅, false 代表默认全部不订阅
  cfg.is_subscribe_full = false;

  // 配置UMS信息:
  //   username/password 账户名/密码, 一个账户只能保持一个连接接入（注意: 如果需要使用委托薄功能，注意账号需要有委托薄功能权限）
  //   ums地址配置:
  //     1) ums地址可以配置1-8个 建议值为2 互为主备, ums_server_cnt 为此次配置UMS地址的个数
  //     2) ums_servers 为UMS地址信息数据结构:
  //        local_ip 为本地地址,填0.0.0.0 或者本机ip
  //        server_ip 为ums服务端地址
  //        server_port 为ums服务端端口
  strncpy(cfg.username, username.c_str(), sizeof(cfg.username));
  strncpy(cfg.password, password.c_str(), sizeof(cfg.password));


  cfg.ums_server_cnt = 2U;
  folly::strlcpy(cfg.ums_servers[0].local_ip, kZeroIp, std::strlen(kZeroIp) + 1U);
  folly::strlcpy(cfg.ums_servers[0].server_ip, server1.data(), server1.length() + 1);
  cfg.ums_servers[0].server_port = static_cast<uint16_t>(port1);

  folly::strlcpy(cfg.ums_servers[1].local_ip, kZeroIp, std::strlen(kZeroIp) + 1U);
  folly::strlcpy(cfg.ums_servers[1].server_ip, server2.data(), server2.length() + 1);
  cfg.ums_servers[1].server_port = static_cast<uint16_t>(port2);

  // 业务数据回调接口(不包括 OnIndicator/OnLog等功能数据回调)的线程安全模式设置:
  //   true: 所有的业务数据接口为接口集线程安全
  //   false: 业务接口单接口为线程安全,接口集非线程安全
  cfg.is_thread_safe = false;

  // 委托薄前档行情参数设置(仅委托薄版本API有效，若非委托薄版本参数设置无效):
  //   1) 行情输出设置，包括档位数量以及每档对应的委托队列数量
  //   2）委托薄计算输出参数设置，包括线程数量以及递交间隔设置
  cfg.enable_order_book = false; //是否开启委托薄计算功能 (注意：委托薄需要订阅相应的逐笔成交和逐笔委托)
  cfg.entry_size        = 10U;   //委托薄输出档位数量
  cfg.order_queue_size  = 10U;   //每个档位输出的委托队列揭示
  cfg.thread_num        = 3U;    //计算委托薄的线程数量

  // 递交的最小时间间隔，默认为0
  cfg.order_book_deliver_interval_microsecond = 0U;

  // 初始化回调以及配置信息,此函数为异步函数,
  // 如果通道初始化以及登陆出现问题会通过onLog/onEvent返回初始化结果信息
  if (IAMDApi::Init(this, cfg) != ErrorCode::kSuccess) {
    XLOG(ERR, "IAMDApi::Init失败");
    return 1;
  }
  XLOG(INFO) << "Init AMA done";

  //订阅示例1
  {
    // 按数据权限订阅信息设置:
    //   1. 订阅信息分三个维度 market:市场, flag:数据类型(比如现货快照, 逐笔成交, 指数快照等), 证券代码
    //   2. 订阅操作有三种:
    //      kSet 设置订阅, 以市场为单位覆盖订阅信息
    //      kAdd 增加订阅, 在前一个基础上增加订阅信息
    //      kDel 删除订阅, 在前一个基础上删除订阅信息
    //      kCancelAll 取消所有订阅信息
    static constexpr uint32_t kSubSize = 1U;
    amd::ama::SubscribeItem   sub[kSubSize];
    std::memset(sub, 0, sizeof(sub));

    // 全部市场
    sub[0].market = 0;
    // 只订阅快照
    sub[0].flag = SubscribeDataType::kSnapshot | SubscribeDataType::kFutureSnapshot | SubscribeDataType::kOptionSnapshot;
    // 全部证券
    sub[0].security_code[0] = '\0';

    /* 发起订阅 */
    auto rc = IAMDApi::SubscribeData(SubscribeType::kSet, &sub[0], kSubSize);
    if (rc != ErrorCode::kSuccess) {
      XLOGF(ERR, "IAMDApi::SubscribeData fails {}", rc);
      return 1;
    }
    XLOG(INFO) << "IAMDApi::SubscribeData success";
  }
  return 0;
}

void AmaInstance::close() {
  if (_closed.exchange(true)) {
    return;
  }

  auto rc = IAMDApi::Release();
  if (rc == 0) {
    XLOG(INFO) << "IAMDApi::Release done";
  } else {
    XLOG(ERR) << "IAMDApi::Release fails " << rc;
  }
}

void AmaInstance::initSnapshot(ItemType& ctx) {
  static uint64_t kLastRecvSeq{0};

  // 时间戳
  uint64_t timestamp{0};

  // 接收总数
  _recv_md_count += ctx.size;

  // 打印间隔,打印后重置为当前接收数量
  if (_recv_md_count - kLastRecvSeq >= FLAGS_print_period) {
    kLastRecvSeq = _recv_md_count;
    timestamp    = folly::hardware_timestamp();
    XLOG(INFO) << "Recv " << _recv_md_count << " MDSnapshots";
  }
}

void AmaInstance::OnMDSnapshot(MDSnapshot* snapshots, uint32_t cnt) {
  ItemPtr pItem(new ItemType(kMdSnapshot, snapshots, cnt));
  initSnapshot(*pItem);
  _queue->write(std::move(pItem));
}

void AmaInstance::OnMDOptionSnapshot(MDOptionSnapshot* snapshots, uint32_t cnt) {
  ItemPtr pItem(new ItemType(kMdOptionSnapshot, snapshots, cnt));
  initSnapshot(*pItem);
  _queue->write(std::move(pItem));
}

void AmaInstance::OnMDIndexSnapshot(MDIndexSnapshot* snapshots, uint32_t cnt) {
  ItemPtr pItem(new ItemType(kMdIndexSnapshot, snapshots, cnt));
  initSnapshot(*pItem);
  _queue->write(std::move(pItem));
}

void AmaInstance::OnMDFutureSnapshot(MDFutureSnapshot* snapshots, uint32_t cnt) {
  ItemPtr pItem(new ItemType(kMdFutureSnapshot, snapshots, cnt));
  initSnapshot(*pItem);
  _queue->write(std::move(pItem));
}

} // namespace tradev::app
